package zmqobj

import (
	"fmt"
	"sync"
	"time"

	"github.com/hjl/db/glog"
	"github.com/hjl/db/utils"
	"github.com/hjl/db/zwrapper/zwrapper"
	zmq "github.com/pebbe/zmq4"
)

// rpc 回调函数
type RpcFunc func(str string)

// rpc 回调信息结构体
type RpcResp struct {
	cb  RpcFunc //函数
	end int     //失效时间
}

// 回调接口，必须实现Do方法
type Worker interface {
	Do(strs []string)
}

// zmqObj结构体
type ZmqObj struct {
	bindSocket *zmq.Socket            // 绑定 socket
	peerSocks  map[string]*zmq.Socket // 对端 socket 缓存
	bindAddr   string                 // 本端绑定 ipc 地址
	msgNum     int                    // 消息编号计数器
	mutex      sync.Mutex             // 并发锁，避免并发写 socket
	updateFreq int                    // 调用 update 函数的频率, 单位毫秒
	recvTimes  int                    // 一次 update 最多调用 recv 的次数
	logger     *glog.ModuleLogger     // 日志对象
	worker     Worker                 // 业务逻辑处理器（用户实现）
	cbInfo     map[string]*RpcResp    // rpc 回调信息，回调函数注册表（msgId -> 回调函数）
}

// 构造一个zmqObj对象
func NewZmqObj(bindAddr, logpath string, worker Worker) *ZmqObj {
	zobj := &ZmqObj{
		peerSocks:  make(map[string]*zmq.Socket),
		bindAddr:   bindAddr,
		msgNum:     0,
		updateFreq: 60,
		recvTimes:  1000,
		logger:     glog.New(logpath),
		worker:     worker,
		cbInfo:     make(map[string]*RpcResp),
	}
	return zobj
}

// 获取绑定地址
func (zobj *ZmqObj) GetBindAddr() string {
	return zobj.bindAddr
}

// 获取绑定 socket
func (zobj *ZmqObj) GetBindSocket() *zmq.Socket {
	return zobj.bindSocket
}

// 获取日志对象
func (zobj *ZmqObj) GetLogger() *glog.ModuleLogger {
	return zobj.logger
}

// 启动 zmq 服务
func (zobj *ZmqObj) Start(gname string) {
	zobj.bindSocket = zwrapper.Bind(zobj) // 创建并绑定 IPC socket；
	if zobj.bindSocket == nil {
		zobj.logger.Error("Start 函数中zmq服务启动失败!!")
		return
	}
	// 移除了注册逻辑（IPC 不需要向中心注册）
	zwrapper.StartServ(zobj, zobj.updateFreq, zobj.recvTimes) // 启动服务监听接收数据（非阻塞）；
}

// 关闭所有 socket
func (zobj *ZmqObj) ClosingAllSocks() {
	zobj.bindSocket.Close()
	for _, v := range zobj.peerSocks {
		v.Close()
	}
}

// 获取对端 socket
func (zobj *ZmqObj) GetPeerSock(addr string) (*zmq.Socket, error) {
	peerSock, ok := zobj.peerSocks[addr]
	if !ok {
		newSocket, err := zmq.NewSocket(zmq.DEALER)
		if err != nil {
			return nil, err
		}
		zobj.peerSocks[addr] = newSocket
		peerSock = newSocket
		err = peerSock.Connect(addr)
		if err != nil {
			return nil, err
		}
	}
	return peerSock, nil
}

// 发送消息 + 注册回调
func (zobj *ZmqObj) Send2Addr(addr, rpcName string, rpcArgs []string, cb RpcFunc) {
	argstr := utils.EncodeLuaTableString(rpcArgs) // 将消息编码成 Lua 表格式字符串

	// 上锁，避免多个 goroutine 同时发消息冲突；
	defer zobj.mutex.Unlock()
	zobj.mutex.Lock()

	// 构造一个唯一 msgId 作为标识；
	zobj.msgNum++
	msgId2str := fmt.Sprintf("%d", zobj.msgNum)
	zobj.logger.Info("[Send2Addr]: msgId2str=%s, addr=%s, rpcName=%s, argstr=%s", msgId2str, addr, rpcName, argstr)

	// 添加 ipc:// 协议前缀
	ipcAddr := "ipc://" + addr

	// 调用 zwrapper.Send(...) 发送消息；
	_, err := zwrapper.Send(zobj, ipcAddr, msgId2str, rpcName, argstr)
	if err != nil {
		zobj.logger.Info("[Send2Addr]: send error, msgId2str=%s, addr=%s, rpcName=%s, argstr=%s, err=%v", msgId2str, addr, rpcName, argstr, err)
	} else {
		// 如果有回调函数 cb，注册到 cbInfo 表中，设置过期时间为 30 秒后
		if cb != nil {
			zobj.cbInfo[msgId2str] = &RpcResp{
				cb:  cb,
				end: time.Now().Second() + 30, // n 秒还没返回 cb 就失效
			}
		}
	}
}

// 接收来自对方的普通 RPC 消息，调用用户注册的 Do() 方法处理
func (zobj *ZmqObj) Recv(funcName, str, addr string) {
	zobj.worker.Do([]string{funcName, str, addr})
}

// 接收响应消息（带回调）
func (zobj *ZmqObj) RespRecv(msgId2str, str string) {
	info, ok := zobj.cbInfo[msgId2str] // 根据 msgId2str 查找回调函数；
	if ok {
		cb := info.cb
		info.cb = nil
		info.end = 0 //强制失效
		go cb(str)   // 调用并清除回调
	}

	//清理过期回调信息
	now := time.Now().Second()
	for k, v := range zobj.cbInfo {
		if v.end < now {
			delete(zobj.cbInfo, k) //如果太多会导致内存泄漏
		}
	}
}

// 接收数据失败时记录日志（例如 socket 断开等异常）
func (zobj *ZmqObj) RecvFail(err error) {
	zobj.logger.Info("[RecvFail]: %v", err)
}
